package com.heima.stream.app;

import com.heima.common.constants.message.HotArticleConstants;
import com.heima.common.util.JsonUtils;
import com.heima.model.article.dtos.ArticleBehaviorMsg;
import com.heima.model.article.dtos.ArticleBehaviorStreamMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.Date;

import static com.heima.common.constants.message.HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC;
import static com.heima.common.constants.message.HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC;

@Slf4j
@EnableBinding(ArticleBehaviorNumApp.ArticleBehaviorNumProcess.class)
public class ArticleBehaviorNumApp {

    @StreamListener(HOT_ARTICLE_SCORE_STREAM_TOPIC)
    @SendTo(HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
    public KStream<String,String> process(KStream<String,String> input){

//        接收的消息 key-null，value-"{"num":1,"articleId":"84848","type":"views"}"
//        转换成 key - “84848”  value-不变
       return  input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                ArticleBehaviorMsg behaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                return new KeyValue<>(behaviorMsg.getArticleId().toString(),value);
            }
        }).
//                设置分组
        groupByKey(Grouped.with(Serdes.String(),Serdes.String())).
//                设置时间窗口
        windowedBy(TimeWindows.of(Duration.ofSeconds(10)).grace(Duration.ZERO)).
//                设置聚合
        aggregate(new Initializer<String>() {
            /**
             * 初始化计算结果
             * @return
             */
            @Override
                public String apply() {
                    log.info("===初始化===");
                    return null;
                }
            }, new Aggregator<String, String, String>() {
                /**
                 * 聚合计算
                 * @param key 本次计算接收到的key
                 * @param value 本次计算接收到的消息值
                 * @param aggregate 上次的计算结果 json字符串
                 * @return
                 */
                @Override
                public String apply(String key, String value, String aggregate) {

                    ArticleBehaviorStreamMsg streamMsg = null;
                    if(StringUtils.isBlank(aggregate)){
//                        如果没有上一次计算结果，就创建计算结果对象
                        streamMsg = new ArticleBehaviorStreamMsg();
                        streamMsg.setArticleId(Long.valueOf(key));
                        streamMsg.setLikes(0L);
                        streamMsg.setViews(0L);
                        streamMsg.setComments(0L);
                        streamMsg.setCollection(0L);
                    }else {
//                        如果有上一次计算结果，就json转对象
                        streamMsg =JsonUtils.toBean(aggregate, ArticleBehaviorStreamMsg.class);
                    }
                    ArticleBehaviorMsg behaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                    switch (behaviorMsg.getType()){
                        case VIEWS:
                            streamMsg.setViews(streamMsg.getViews() + behaviorMsg.getNum());
                            break;
                        case LIKES:
                            streamMsg.setLikes(streamMsg.getLikes() + behaviorMsg.getNum());
                            break;
                        case COMMENT:
                            streamMsg.setComments(streamMsg.getComments() + behaviorMsg.getNum());
                            break;
                        case COLLECTION:
                            streamMsg.setCollection(streamMsg.getCollection() + behaviorMsg.getNum());
                            break;
                    }

                    return JsonUtils.toString(streamMsg);
                }
           },Materialized.with(Serdes.String(),Serdes.String())).
                suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())).
                toStream().
                map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        log.info("time={},key={},value={}",new Date(key.window().start()),key.key(),value);
                        return new KeyValue<>(key.key(),value);
                    }
                });
    }


    public interface ArticleBehaviorNumProcess{

        @Input(HOT_ARTICLE_SCORE_STREAM_TOPIC)
        KStream<String,String> input();
        @Output(HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
        KStream<String,String> output();
    }
}
